/*
 * Copyright (c) 2010 Vadim Zhukov <persgray@gmail.com>
 *
 * Permission to use, copy, modify, and distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
 * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */


#include "common.h"

#include <sys/socket.h>
#include <sys/un.h>
#include <arpa/nameser.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#include <assert.h>
#include <err.h>
#include <poll.h>
#include <resolv.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include "../assl/assl.h"
#include "logmsg.h"
#include "peertask.h"
#include "peer.h"
#include "resolver.h"
#include "sensor.h"
#include "syslogd.h"
#include "timestamp.h"


#define DNS_CHECK_IN		60	/* Check DNS for collectors list */
                                        /* changes each N sec */
#define LOG_POLL_TIME		1	/* Check log buffers each X secs */


extern char			*__progname;
int				 debug = 0;
static volatile sig_atomic_t	 quit_requested = 0;
static volatile sig_atomic_t	 dump_requested = 0;
static struct peer_task_queue	 unsent_lines =
    SIMPLEQ_HEAD_INITIALIZER(unsent_lines);
static struct wsensor_list	 sensors = TAILQ_HEAD_INITIALIZER(sensors);
static struct watchdog_list	 watchdogs = TAILQ_HEAD_INITIALIZER(watchdogs);


static void		 process_sensors(void);
static int		 handle_failed_line(struct peer *peer,
			    struct peer_task *);
static int		 handle_conf_change(struct peer *peer,
			    struct peer_task *);
static uint32_t		 check_session(struct peer_list *, const char *,
			    uint32_t, int *);
static struct pollfd	*build_poll_pool(struct wsensor_list *,
			    struct peer_list *, size_t *);
static __dead void	 usage(const char *);
static void		 sighandler(int);
static void		 dump_unsent_lines(void);


struct peer_task_handler	task_handlers[] = {
{ &handle_failed_line, PEER_QUEUE_OUTPUT, PEER_CMD_LINE,         PEER_TH_F_FAIL    },
{ &handle_conf_change, PEER_QUEUE_INPUT,  PEER_CMD_CONF_CHANGED, PEER_TH_F_SUCCESS },
};


/*
 * Run through the sensors, feed their data to watchdogs and make up log message
 * tasks for sending to collector finally.
 */
void
process_sensors(void) {
	struct wsensor		*s;
	struct watchdog		*w;
	struct log_msg		*m, *mt;
	struct peer_task	*task;

	TAILQ_FOREACH(s, &sensors, entry) {
		DPRINTFX(3, "process_sensors: sensor: %s", s->name);
		while ((m = sensor_get_line(s))) {
			TAILQ_FOREACH(w, &watchdogs, entry)
				watchdog_check(w, s, m);
			for (; m != NULL; m = mt) {
				mt = m->next;
				if (!(m->hdr->flags & LOG_MSG_ACTION_LOG))
					goto freem;
				task = log_msg_to_peer_task(m);
				if (task == NULL) {
					DPRINTF(1, "log_msg_to_peer_tasks");
					goto freem;
				}
				SIMPLEQ_INSERT_TAIL(&unsent_lines, task, entry);
freem:
				log_msg_free(m);
			}
			if (sensor_is_gauge(s))
				break;
		}
	}
}

int
handle_failed_line(struct peer *peer, struct peer_task *task) {
	SIMPLEQ_INSERT_HEAD(&unsent_lines, task, entry);
	return PEER_TH_CB_CARED;
}

int
handle_conf_change(struct peer *peer, struct peer_task *task) {
	struct peer_msg_conf_changed	*pmcc;
	struct wsensor			*s;
	struct watchdog			*w;
	int				 oerrno, watchdog_op;

	assert(is_conf_task(task));
	w = NULL;
	s = NULL;
	if (task->cmd.pmh_bufcount < 2) {
		DPRINTFX(1, "handle_conf_change: not enough data buffers");
		goto taskerr;
	}
	if (task->databuf[1]->len > 7 &&
	    strncmp(task->databuf[1]->data, "sensor/", 7))
		watchdog_op = 0;
	else if (task->databuf[1]->len > 9 &&
	    strncmp(task->databuf[1]->data, "watchdog/", 9))
		watchdog_op = 1;
	else {
		DPRINTFX(1, "handle_conf_change: unknown object to configure: "
		    "%s", task->databuf[1]->data);
		goto taskerr;
	}
	pmcc = (struct peer_msg_conf_changed *)task->databuf[0]->data;
	switch (pmcc->op) {
	case PEER_CONF_OP_ADDOBJ:
		if (watchdog_op) {
			if (task->cmd.pmh_bufcount < 3) {
				DPRINTFX(1, "handle_conf_change: not enough "
				    "data buffers");
				goto taskerr;
			}
			w = watchdog_new();
			if (w == NULL)
				goto err;
			if (strlcpy(w->name, task->databuf[1]->data,
			    sizeof(w->name)) >= sizeof(w->name)) {
				DPRINTFX(1, "handle_conf_change: object name "
				    "too long: %s", task->databuf[1]->data);
				goto taskerr;
			}
			if (watchdog_from_spec(w,
			    task->databuf[2]->data) == -1)
				goto taskerr;
			TAILQ_INSERT_TAIL(&watchdogs, w, entry);
		} else {
			s = malloc(sizeof(struct wsensor));
			if (s == NULL)
				goto err;
			if (strlcpy(s->name, task->databuf[1]->data,
			    sizeof(s->name)) >= sizeof(s->name)) {
				DPRINTFX(1, "handle_conf_change: object name "
				    "too long: %s", task->databuf[1]->data);
				goto taskerr;
			}
			if (sensor_init(s) == -1)
				goto taskerr;
			TAILQ_INSERT_TAIL(&sensors, s, entry);
		}
		break;

	case PEER_CONF_OP_DELOBJ:
		if (watchdog_op) {
			TAILQ_FOREACH(w, &watchdogs, entry) {
				if (!strcmp(w->name, task->databuf[1]->data)) {
					TAILQ_REMOVE(&watchdogs, w, entry);
					watchdog_free(w);
					goto done;
				}
			}
			errno = ESRCH;
			goto err;
		} else {
			TAILQ_FOREACH(s, &sensors, entry) {
				if (!strcmp(s->name, task->databuf[1]->data)) {
					TAILQ_REMOVE(&sensors, s, entry);
					sensor_uninit(s);
					free(s);
					goto done;
				}
			}
			errno = ESRCH;
			goto err;
		}
		break;

	default:
		DPRINTFX(1, "handle_conf_change: illegal conf change operation "
		    "code %d", (int)pmcc->op);
		goto taskerr;
	}

done:
	return PEER_TH_CB_FREE;

err:
	oerrno = errno;
	free(s);
	free(w);
	DPRINTF(0, "handle_conf_change");
	errno = oerrno;
	return PEER_TH_CB_FREE;

taskerr:
	free(s);
	free(w);
	DPRINTFX(2, "handle_conf_change: dropping connection");
	peer_close(peer);
	errno = EINVAL;
	return PEER_TH_CB_FREE;
}

/*
 * Try to find current session number. If no peer sessions are running, try to
 * establish session with any collector.
 *
 * Returns session number if session found, or cursession value if failed.
 *
 * Always set nsessions to correct value, even if fails.
 */
uint32_t
check_session(struct peer_list *peers, const char *myname, uint32_t cursession,
    int *nsessions) {
	static struct	peer lastpeer;
	struct		peer *peer, *speer;

	*nsessions = 0;
	if (TAILQ_EMPTY(peers)) {
		bzero(&lastpeer, sizeof(struct peer));
		return 0;
	}

	/* At first, proceed queues & check if session is already established */
	TAILQ_FOREACH(peer, peers, entry) {
		peer_run_queues(peer);
		if (peer_session_established(peer)) {
			cursession = max(cursession, peer->session);
			(*nsessions)++;
		}
	}
	/* check if lastpeer is still actual */
	for (peer = TAILQ_FIRST(peers); peer != NULL;
	    peer = TAILQ_NEXT(peer, entry))
		if (!compare_peers(peer, &lastpeer))
			break;
	speer = peer;
retry:
	if (peer == NULL)
		peer = TAILQ_FIRST(peers);
	if (peer_session_established(peer)) {
		(*nsessions)++;
		return peer->session;
	}
	if (!peer_connected(peer))
		if (peer_connect(peer) == -1)
			goto fail;
	if (!peer_hello_sent(peer))
		if (!peer_start_session(peer, cursession, myname))
			goto fail;
	peer_run_queues(peer);
	if (peer_hello_sent(peer)) {
		/* Now just wait */
		memcpy(&lastpeer, peer, sizeof(struct peer));
		return 0;
	}

fail:
	peer = TAILQ_NEXT(peer, entry);
	if (peer == speer) {
		/* The circle is over */
		if (peer == NULL)
			bzero(&lastpeer, sizeof(struct peer));
		else
			memcpy(&lastpeer, peer, sizeof(struct peer));
		return 0;
	}
	goto retry;
}

/*
 * Return an array of pollfd strctures, and its size is placed in count.
 * If there are no connected peers, returns NULL.
 */
struct pollfd *
build_poll_pool(struct wsensor_list *sensors, struct peer_list *peers,
    size_t *count) {
	struct	peer *peer;
	struct	pollfd *pool;
	struct	wsensor *s;
	size_t	i;

	i = 0;
	TAILQ_FOREACH(s, sensors, entry)
		if (sensor_is_pollable(s))
			i++;
	TAILQ_FOREACH(peer, peers, entry)
		if (peer_connected(peer))
			i++;
	*count = i;
	if (i == 0)
		return NULL;
	pool = calloc(sizeof(struct pollfd), *count);
	if (pool == NULL)
		fatalx("build_poll_pool: calloc");
	i = 0;
	TAILQ_FOREACH(s, sensors, entry)
		if (sensor_is_pollable(s)) {
			pool[i].fd = s->fd;
			pool[i].events = POLLIN;
			i++;
		}
	TAILQ_FOREACH(peer, peers, entry)
		if (peer_connected(peer)) {
			pool[i].fd = peer_get_fd(peer);
			pool[i].events = POLLIN | POLLOUT;
			peer->pollptr = &pool[i];
			i++;
		} else
			peer->pollptr = NULL;
	return pool;
}

void
usage(const char *msg) {
	if (msg != NULL)
		fprintf(stderr, "%s: %s", __progname, msg);
	fprintf(stderr,
"usage: %s [-d] [-P peer] [-p peerdomain] [-S sensor] [-s socket]\n"
"    [-t rrtype] [-W watchdog]\n"
"  -d             Do not daemonize and increase verbosity.\n"
"  -P peer        Force connection to specified peer. Can be specified more than\n"
"                 once, connections will be established with all peers.\n"
"  -p peerdomain  DNS name to resolve for collectors. Defaults to\n"
"                 \"" DEF_PEER_DNS_PREFIX "\" plus current domain name\n"
"                 (i.e. full host name without first part). Cannot be combined\n"
"                 with -f option.\n"
"  -S sensor      Add a sensor with specified name.\n"
"  -s socket      Path to syslogd socket, defaults to " DEF_SYSLOG_SOCK ".\n"
"  -t rrtype      Resource record type to resolve (A or SRV). Note: SRV\n"
"                 records are not supported yet - defaults to A until.\n"
"  -W watchdog    Add a watchdog. Watchdog specification as follows:.\n"
"                 WATCHDOG: name,if:COND=CONDVALUE[,...],do:ACTION=PARAMS[,...]\n"
"                 COND: SENSOR+OP\n"
"                 CONDVALUE: VALUE[+VALUE]\n"
"                 PARAMS: PARAM[+PARAM]\n"
"                 PARAM: NAME=VALUE\n"
"\n"
"Peer addresses could be specified in either form:\n"
"  IPv4:  W.X.Y.Z  or  W.X.Y.Z:port\n"
"  IPv6:  A:B:C:D:E:F::J:K  or  [A:B:C:D:E:F::J:K]  or  [A:B:C:D:E:G::J:K]:port\n",
	    __progname);
	exit(EXIT_FAILURE);
}

void
sighandler(int sig) {
	switch (sig) {
#ifdef HAS_SIGINFO
	case SIGINFO:
#endif
	case SIGUSR1:
		dump_requested = 1;
		break;

	case SIGHUP:		/* Not implemented yet */
	case SIGPIPE:		/* We handle those errors ourself */
	case SIGUSR2:		/* Not implemented yet */
		break;

	default:
		quit_requested = 1;
	}
}

void
dump_unsent_lines(void) {
	struct peer_task	*task;

	fputs("\n===> Dumping unsent lines <===\n", stderr);
	SIMPLEQ_FOREACH(task, &unsent_lines, entry)
		peer_task_dump(stderr, task, 1);
}

int
main(int argc, char **argv) {
	struct addrinfo		*colladdrs, *newcolladdrs;
	struct bindaddr		 ba;
	struct peer_list	 collectors;
	struct peer		*peer;
	struct wsensor		*s;
	struct watchdog		*w;
	struct sigaction	 siga;
	struct pollfd		*pool;
	struct ssl_paths	 ssl_paths;
	size_t			 poolsize;
	time_t			 t, lastdns;
	uint32_t		 session;
	int			 c, n, fixedpeers, peer_sessions, rrtype;
	int			 pollonly;
	char			*peerdomain;
	char			 myname[MAXHOSTNAMELEN];

	strlcpy(ssl_paths.ca_crt, CERTPATH "/ca/ca.crt",
	    sizeof(ssl_paths.ca_crt));
	strlcpy(ssl_paths.my_crt, CERTPATH "/server/server.crt",
	    sizeof(ssl_paths.ca_crt));
	strlcpy(ssl_paths.my_key, CERTPATH "/server/private/server.key",
	    sizeof(ssl_paths.ca_crt));

	bzero(myname, MAXHOSTNAMELEN);
	TAILQ_INIT(&collectors);

	assl_initialize();
	gethostname(myname, MAXHOSTNAMELEN);

	session = 0;
	peer_sessions = 0;
	colladdrs = NULL;
	newcolladdrs = 0;
	peerdomain = NULL;
	rrtype = -1;
	fixedpeers = 0;
	pollonly = 1;
	while ((c = getopt(argc, argv, "dP:p:S:s:t:")) != -1) {
		switch (c) {
		case 'd':
			debug++;
			break;

		case 'P':
			fixedpeers = 1;
			if (peerdomain != NULL)
				usage("-P and -p are mutally exclusive");
			if (rrtype != -1)
				usage("-P and -t are mutally exclusive");
			n = strlen(optarg);
			if (n == 0)
				fatalx("empty peer address supplied");
			peer = peer_new(0, NULL);
			if (peer == NULL)
				fatal("peer_new");
			if (fill_sa_from_user(optarg, &ba,
			    SASPEC_HOST_NUMERIC, PEER_PORT) == -1)
				fatal("setting up peer address %s", optarg);
			peer->addrlen = ba.slen;
			memcpy(&peer->addr, &ba.ss, sizeof(peer->addr));
			peer->af = peer->addr.ss_family;
			peer->ssl_paths = &ssl_paths;
			peer->status = PEER_STATUS_NEW;
			TAILQ_INSERT_TAIL(&collectors, peer, entry);
			break;

		case 'p':
			if (peerdomain != NULL)
				usage("-p cannot be used twice");
			if (fixedpeers)
				usage("-P and -p are mutally exclusive");
			peerdomain = strdup(optarg);
			if (peerdomain == NULL)
				fatal("strdup");
			break;

		case 'S':
			s = malloc(sizeof(struct wsensor));
			if (s == NULL)
				fatalx("malloc(sensor)");
			bzero(s, sizeof(struct wsensor));
			if (strlcpy(s->name, optarg, sizeof(s->name)) >=
			    sizeof(s->name)) {
				fatalx("sensor name is too long: %s",
				    optarg);
			}
			if (sensor_init(s) == -1)
				fatal("sensor_init");
			if (!sensor_is_pollonly(s))
				pollonly = 0;
			TAILQ_INSERT_TAIL(&sensors, s, entry);
			break;

		case 's':
			if (sl_sockpath.sun_path[0] != '\0')
				usage("-s cannot be used twice");
			n = strlen(optarg);
			if (n == 0)
				usage("socket address is empty");
			if (n > sizeof(sl_sockpath.sun_path))
				usage("socket address is too long");
			strncpy(sl_sockpath.sun_path, optarg,
			    sizeof(sl_sockpath.sun_path));
			break;

		case 't':
			if (rrtype != -1)
				usage("-t cannot be used twice");
			if (fixedpeers)
				usage("-P and -t are mutally exclusive");
			if (strcmp(optarg, "A"))
				rrtype = T_A;
			else if (strcmp(optarg, "SRV"))
				rrtype = T_SRV;
			else
				usage("unsupported resource record type in -t");
			break;

		case 'W':
			w = watchdog_new();
			if (w == NULL)
				fatal("watchdog_new");
			if (watchdog_from_spec(w, optarg) == -1)
				fatalx("watchdog_from_spec");
			TAILQ_INSERT_TAIL(&watchdogs, w, entry);
			break;

		default:
			usage(NULL);
		}
	}
	argc -= optind;
	argv += optind;
	if (argc > 0)
		usage("extra parameters in command line");

	DPRINTFX(2, "pollonly is %d", pollonly);

	if (rrtype == -1)
		rrtype = T_A;

	if (!fixedpeers) {
		colladdrs = find_collectors(peerdomain, rrtype);
		if (colladdrs == NULL)
			DPRINTFX(1, "could not find any collectors");
	}
	peer_update_connections(&collectors, colladdrs, PEER_CONN_PLAIN,
	    &peer_sessions, NULL, &ssl_paths);
	pool = build_poll_pool(&sensors, &collectors, &poolsize);

	bzero(&siga, sizeof(struct sigaction));
	siga.sa_handler = sighandler;
	sigaction(SIGHUP, &siga, NULL);
	sigaction(SIGINT, &siga, NULL);
	sigaction(SIGPIPE, &siga, NULL);
	sigaction(SIGTERM, &siga, NULL);
	sigaction(SIGUSR1, &siga, NULL);
	sigaction(SIGUSR2, &siga, NULL);
#ifdef HAS_SIGINFO
	sigaction(SIGINFO, &siga, NULL);
#endif

	if (!debug && daemon(0, 0))
		DPRINTF(0, "could not go daemon");

	lastdns = time(NULL);
	for (;;) {
		t = time(NULL);
		if (pool == NULL) {
			DPRINTFX(4, "calling process_sensors()");
			process_sensors();
			goto checks;
		}

		DPRINTFX(4, "calling poll()");
		n = poll(pool, poolsize,
		    pollonly ? INFTIM : (LOG_POLL_TIME * 1000));
		if (n == -1) {
			if (errno != EINTR)
				DPRINTF(0, "poll");
			goto checks;
		}

		DPRINTFX(4, "calling process_sensors()");
		process_sensors();
		if (n == 0)
			goto checks;

		session = check_session(&collectors, myname, session,
		    &peer_sessions);
		/* XXX: should peer_sessions check be enough? */
		if (session == 0 || peer_sessions == 0)
			goto checks;
		peer_send_tasks_to_any(&unsent_lines, &collectors);
		TAILQ_FOREACH(peer, &collectors, entry) {
			if (peer->pollptr == NULL)
				continue;
			if (peer->pollptr->revents)
				n--;
			peer_handle(peer, task_handlers,
			    nitems(task_handlers));
			if (n == 0)
				break;
		}

checks:
		if (dump_requested) {
			dump_unsent_lines();
			dump_requested = 0;
		}

		if (quit_requested) {
			DPRINTFX(2, "exit signal sent, so exiting");
			break;
		}

		if (pool == NULL)
			sleep(LOG_POLL_TIME);

		if (!fixedpeers && t - lastdns >= DNS_CHECK_IN) {
			DPRINTFX(4, "calling find_collectors()");
			newcolladdrs = find_collectors(peerdomain, rrtype);
			if (newcolladdrs != NULL) {
				freeaddrinfo(colladdrs);
				colladdrs = newcolladdrs;
			} else
				DPRINTFX(1, "could not find any collectors");
			lastdns = t;
		}
		DPRINTFX(4, "calling peer_update_connections()");
		if (peer_update_connections(&collectors, colladdrs,
		    PEER_CONN_PLAIN, &peer_sessions, NULL, &ssl_paths)) {
			free(pool);
			pool = build_poll_pool(&sensors, &collectors,
			    &poolsize);
		}
	}

	while ((peer = TAILQ_FIRST(&collectors))) {
		TAILQ_REMOVE(&collectors, peer, entry);
		peer_free(peer);
	}
	free(pool);
	while ((w = TAILQ_FIRST(&watchdogs)) != NULL) {
		TAILQ_REMOVE(&watchdogs, w, entry);
		watchdog_free(w);
	}
	while ((s = TAILQ_FIRST(&sensors)) != NULL) {
		TAILQ_REMOVE(&sensors, s, entry);
		sensor_uninit(s);
		free(s);
	}
	if (colladdrs != NULL)
		freeaddrinfo(colladdrs);

	return (0);
}
